package org.apache.rocketmq.proxy.service.route;

import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.google.common.base.Optional;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory;
import org.apache.rocketmq.client.latency.MQFaultStrategy;
import org.apache.rocketmq.client.latency.Resolver;
import org.apache.rocketmq.client.latency.ServiceDetector;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.Address;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.utils.AbstractStartAndShutdown;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
import org.apache.rocketmq.proxy.service.route.TopicRouteHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public abstract class TopicRouteService extends AbstractStartAndShutdown {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    //MQ客户端实例工厂，用于提供MQ客户端API
    private final MQClientAPIFactory mqClientAPIFactory;
    //忽略这个成员变量，在第八版本代码中用不到
    private MQFaultStrategy mqFaultStrategy;

    //Proxy模块专门缓存从Namesrv获取的主题路由信息
    protected final LoadingCache<String/*topic*/, MessageQueueView> topicCache;


    //这个成员变量在第八版本代码中也没用上
    protected final ScheduledExecutorService scheduledExecutorService;

    protected final ThreadPoolExecutor cacheRefreshExecutor;


    public TopicRouteService(MQClientAPIFactory mqClientAPIFactory) {
        //得到Proxy模块的配置信息
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        this.mqClientAPIFactory = mqClientAPIFactory;
        this.scheduledExecutorService = ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
                "TopicRouteService_"));

        this.cacheRefreshExecutor = ThreadPoolMonitor.createAndMonitor(
                config.getTopicRouteServiceThreadPoolNums(),
                config.getTopicRouteServiceThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                "TopicRouteCacheRefresh",
                config.getTopicRouteServiceThreadPoolQueueCapacity()
        );

        this.topicCache = Caffeine.newBuilder()
                //设置缓存最大容量，超过此容量将按照LRU策略对旧数据进行清理
                .maximumSize(config.getTopicRouteServiceCacheMaxNum())
                //设置缓存项被访问后多久过期的时间
                .expireAfterAccess(config.getTopicRouteServiceCacheExpiredSeconds(), TimeUnit.SECONDS)
                //设置缓存项的过期时间
                .refreshAfterWrite(config.getTopicRouteServiceCacheRefreshSeconds(), TimeUnit.MILLISECONDS)
                .executor(cacheRefreshExecutor)
                //构建缓存对象，这里在构建的时候创建了一个CacheLoader对象
                //在缓存未命中时，就会通过该加载器的load方法从Namesrv加载指定主题的路由信息
                .build(new CacheLoader<>() {

                    @Override
                    public @Nullable MessageQueueView load(@NonNull String topic) throws Exception {

                        try {
                            TopicRouteData topicRouteData = mqClientAPIFactory.getClient().getTopicRouteInfoFromNameServer(topic,
                                    Duration.ofSeconds(3).toMillis());
                            //构建消息队列视图
                            return buildMessageQueueView(topic, topicRouteData);
                        } catch (Exception e) {
                            log.warn(String.format("get topic route from namesrv. topic: %s", topic), e);
                            if (TopicRouteHelper.isTopicNotExistError(e)) {
                                return MessageQueueView.WRAPPED_EMPTY_QUEUE;
                            }
                            throw e;
                        }
                    }


                    //缓存项刷新方法，当缓存项过期时，会调用此方法来刷新缓存项
                    @Override
                    public @Nullable MessageQueueView reload(@NonNull String key,
                                                             @NonNull MessageQueueView oldValue) throws Exception {
                        try {
                            return load(key);
                        } catch (Exception e) {
                            //加载失败则返回旧值
                            log.warn(String.format("reload topic route from namesrv. topic: %s", key), e);
                            return oldValue;
                        }
                    }
                });

        //定义一个Broker服务探测器，这个探测器会判断指定的Broker节点是否可以访问
        ServiceDetector serviceDetector = new ServiceDetector() {

            //检测指定Broker节点是否可以访问的方法，方法参数endpoint就是指定的Broker节点的网络地址
            //timeoutMillis就是请求超时时间，返回true表示broker节点可以正常访问，返回false表示broker节点可能存在故障，无法访问
            @Override
            public boolean detect(String endpoint, long timeoutMillis) {
                Optional<String> candidateTopic = pickTopic();
                if (!candidateTopic.isPresent()) {
                    return false;
                }
                try {
                    GetMaxOffsetRequestHeader requestHeader = new GetMaxOffsetRequestHeader();
                    requestHeader.setTopic(candidateTopic.get());
                    requestHeader.setQueueId(0);
                    Long maxOffset = mqClientAPIFactory.getClient().getMaxOffset(endpoint, requestHeader, timeoutMillis).get();
                    return true;
                } catch (Exception e) {
                    return false;
                }
            }
        };
        //创建MQFaultStrategy实例，用于实现消息发送的故障转移策略
        //MQFaultStrategy能够选择正确的目标broker，用来发送消息
        mqFaultStrategy = new MQFaultStrategy(
                //从proxy配置信息中提取客户端相关配置，这些配置信息包含发送延迟启用状态、探测器启用状态、探测超时时间，探测间隔时间等等
                extractClientConfigFromProxyConfig(config),
                //创建Resolver解析器对象，用来将broker名称解析为实际的网络地址
                new Resolver() {
                    @Override

                    //解析Broker名称为网络地址的方法
                    public String resolve(String name) {
                        try {
                            //下面的代码其实根本没什么作用，别忘了现在是在proxy本地模式下，在这个模式下
                            //proxy和broker节点是部署在同一进程的
                            ProxyContext ctx = ProxyContext.createForInner("MQFaultStrategy");
                            //而在本地模式下，getBrokerAddr方法返回的就是当前进程中的Broker节点的网路地址
                            //由此可见，这个mqFaultStrategy对象在proxy本地模式下其实并不会发挥作用，只有在proxy集群模式下才会发挥作用
                            //源码中也正是这么做的，所以在第十二版本代码中，大家没必要把注意力放在mqFaultStrategy对象中，先混个眼熟就行
                            String brokerAddr = getBrokerAddr(ctx, name);
                            //返回解析得到的broker地址
                            return brokerAddr;
                        } catch (Exception e) {
                            return null;
                        }
                    }
                },
                serviceDetector
        );
        this.init();
    }

    // pickup one topic in the topic cache
    private Optional<String> pickTopic() {
        if (topicCache.asMap().isEmpty()) {
            return Optional.absent();
        }
        return Optional.of(topicCache.asMap().keySet().iterator().next());
    }

    protected void init() {
        this.appendShutdown(this.scheduledExecutorService::shutdown);
        this.appendStartAndShutdown(this.mqClientAPIFactory);
    }
    @Override
    public void shutdown() throws Exception {

    }

    @Override
    public void start() throws Exception {

    } public ClientConfig extractClientConfigFromProxyConfig(ProxyConfig proxyConfig) {
        ClientConfig tempClientConfig = new ClientConfig();
        tempClientConfig.setSendLatencyEnable(proxyConfig.getSendLatencyEnable());
        tempClientConfig.setStartDetectorEnable(proxyConfig.getStartDetectorEnable());
        tempClientConfig.setDetectTimeout(proxyConfig.getDetectTimeout());
        tempClientConfig.setDetectInterval(proxyConfig.getDetectInterval());
        return tempClientConfig;
    }

    public abstract MessageQueueView getCurrentMessageQueueView(ProxyContext ctx, String topicName) throws Exception;

    public abstract ProxyTopicRouteData getTopicRouteForProxy(ProxyContext ctx, List<Address> requestHostAndPortList,
                                                              String topicName) throws Exception;

    public abstract String getBrokerAddr(ProxyContext ctx, String brokerName) throws Exception;
    public MQFaultStrategy getMqFaultStrategy() {
        return this.mqFaultStrategy;
    }
    protected static boolean isTopicRouteValid(TopicRouteData routeData) {
        return routeData != null && routeData.getQueueDatas() != null && !routeData.getQueueDatas().isEmpty()
                && routeData.getBrokerDatas() != null && !routeData.getBrokerDatas().isEmpty();
    } public MessageQueueView getAllMessageQueueView(ProxyContext ctx, String topicName) throws Exception {
        return getCacheMessageQueueWrapper(this.topicCache, topicName);
    }

    private MessageQueueView getCacheMessageQueueWrapper(LoadingCache<String, MessageQueueView> topicCache, String topicName) throws MQClientException {
        MessageQueueView messageQueueView = topicCache.get(topicName);
        if (messageQueueView != null && messageQueueView.isEmptyCachedQueue()){
            throw new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "topic not exist"+topicName);
        }
        return messageQueueView;
    }

    private MessageQueueView buildMessageQueueView(String topic, TopicRouteData topicRouteData) {
        if (isTopicRouteValid(topicRouteData)) {
            MessageQueueView messageQueueView = new MessageQueueView(topic, topicRouteData, TopicRouteService.this.mqFaultStrategy);
            return messageQueueView;
        }
        return MessageQueueView.WRAPPED_EMPTY_QUEUE;
    }


    //在这个方法内部，会把统计信息交给故障转移组件处理
    public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation,
                                boolean reachable) {
        checkSendFaultToleranceEnable();
        this.mqFaultStrategy.updateFaultItem(brokerName, currentLatency, isolation, reachable);
    }
    public void checkSendFaultToleranceEnable() {
        boolean hotLatencySwitch = ConfigurationManager.getProxyConfig().isSendLatencyEnable();
        boolean hotDetectorSwitch = ConfigurationManager.getProxyConfig().isStartDetectorEnable();
        this.mqFaultStrategy.setSendLatencyFaultEnable(hotLatencySwitch);
        this.mqFaultStrategy.setStartDetectorEnable(hotDetectorSwitch);
    }
}
